package com.gxljc.bear.crawler.weixin;

import com.google.gson.Gson;
import com.google.gson.reflect.TypeToken;
import com.gxljc.commons.util.StringUtil;
import com.gxljc.bear.crawler.base.DolphinFetchData;
import com.gxljc.bear.crawler.proxy.ProxyUtil;
import com.gxljc.bear.crawler.util.RedisUtil;
import com.gxljc.bear.crawler.util.SparkUtil;
import com.gxljc.bear.crawler.util.UrlUtil;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang.StringUtils;
import org.apache.log4j.Logger;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.FlatMapFunction;

import java.io.Serializable;
import java.util.*;

/**
 * Created by jianzhou on 2016/11/24.
 */

public class PcSeedCrawler implements Serializable {
    public static Logger LOG = Logger.getLogger(PcSeedCrawler.class);
    private Boolean proxy = false;
    private String channelType = "";

    public PcSeedCrawler(Boolean proxy) {
        this.proxy = proxy;
    }

    public List<Integer> getSeed() throws Exception {
        Long totalNum = RedisUtil.llen(WeixinConst.WEIXIN_MOBILE_REDIS_KEY_HIS_URLS);
        List<Integer> seeds = new LinkedList<Integer>();
        for (int i = 0; i < totalNum; ++i) {
            seeds.add(i);
        }
        return seeds;
    }

    public void crawl() throws InterruptedException {
        final long SLEEP_TIME = 13 * 60 * 1000;
        while (true) {
            List<Integer> seeds = null;
            try {
                seeds = getSeed();
            } catch (Exception er) {
                er.printStackTrace();
            }

            if (CollectionUtils.isEmpty(seeds)) {
                LOG.error("page is empty, sleep 13 min");
                Thread.sleep(SLEEP_TIME); //没有数据，暂停5分钟
                continue;
            }
            int size = seeds.size();
            ProxyUtil.initQueue(proxy);
            LOG.info("weixin seeds size = " + size);
            if (size < WeixinConst.MAX_SINGLE_CORE) { //少于一定数量时，使用单线程爬取
                try {
                    crawlNoSpark(seeds);
                } catch (Exception er) {
                    er.printStackTrace();
                    LOG.error("single core job error", er);
                }
            } else {
                try {
                    crawl(seeds);
                } catch (Exception er) {
                    er.printStackTrace();
                    LOG.error("spark job error", er);
                }
            }
            LOG.info("sleep 1 min");
            Thread.sleep(60 * 1000); //1分钟检查一次
        }
    }

    public void crawl(List<Integer> seeds) throws Exception {
        int coreMax = 10;
        JavaSparkContext jsc = SparkUtil.createCommonsSparkContext(
                "bear-手机微信种子爬虫-" + seeds.size(), coreMax,
                coreMax * 2, PcPageCrawler.class);
        JavaRDD<Integer> seedsRDD = jsc.parallelize(new ArrayList<Integer>(seeds));
        long count = seedsRDD.mapPartitions(
                new FlatMapFunction<Iterator<Integer>, Integer>() {
                    private static final long serialVersionUID = 1L;

                    @Override
                    public Iterable<Integer> call(Iterator<Integer> seeds)
                            throws Exception {
                        int successCnt = 0;
                        while (seeds.hasNext()) {
                            Integer seed = seeds.next();
                            try {
                                int ret = crawlOnce();
                                if (ret == 1)
                                    successCnt++;
                                if (successCnt % 100 == 0) {
                                    LOG.info("successCnt=" + successCnt);
                                }
                            } catch (Exception er) {
                                er.printStackTrace();
                            }
                        }
                        return Arrays.asList(successCnt);
                    }
                }).count();
        LOG.info("all count=" + count);
        jsc.stop();
    }

    public void crawlNoSpark(List<Integer> seeds) {
        if (CollectionUtils.isEmpty(seeds)) {
            LOG.error("page is empty");
            return;
        }
        for (Integer table : seeds) {
            try {
                crawlOnce();
            } catch (Exception e) {
                e.printStackTrace();  //To change body of catch statement use File | Settings | File Templates.
            }
        }
    }

    public int crawlOnce() throws Exception {
        String url = getHisUrl();
        if (StringUtils.isEmpty(url)) throw new Exception("url is null");
        LOG.info(String.format("read url from redis key: url: %s", url));
        boolean ok = crawlSingle(url);
        if (!ok) {
            LOG.error(String.format("error, crawl history article of url: %s", url));
            return 0;
        }
        return 1;
    }

    /**
     * 单条历史信息列表爬取
     *
     * @param url
     * @return
     */
    public boolean crawlSingle(String url) {
        try {
            String html = getUrlBody(url);
            List<PageTable> res = parseHistoryList(html);
            if (res == null || res.size() == 0) {
                LOG.info(String.format("can not find history articls, url: %s", url));
                return false;
            } else {
                for (PageTable table : res) {
                    LOG.info(table);
                }
                try {
                    // save to mongodb.
                    WeixinMongodbUtil.savePageMongodb(res, WeixinConst.WEIXIN_MOBILE_PAGE);
                    return true;
                } catch (Exception ex) {
                    LOG.error(String.format("save to monodb error:%s", ex));
                    return false;
                }
            }
        } catch (Exception ex) {
            ex.printStackTrace();
            LOG.error(String.format("get all biz map error:%s", ex));
            return false;
        }
    }

    public String getUrlBody(String url) {
        try {
            DolphinFetchData fetcher = DolphinFetchData.getInstance(channelType);
            byte[] html = fetcher.getHtml(url, proxy);
            return new String(html);
        } catch (Exception ex) {
            LOG.error(String.format("error, get url Body failed: %s", ex));
            return "";
        }
    }

    public String getHisUrl() {
        try {
            return RedisUtil.rpop(WeixinConst.WEIXIN_MOBILE_REDIS_KEY_HIS_URLS);
        } catch (Exception ex) {
            return null;
        }
    }

    public List<PageTable> parseHistoryList(String html) {
        String js = findJsonContent(html);
        if (StringUtils.isEmpty(js)) {
            LOG.info(String.format("can not find JsonContent: "));
            return null;
        }
        Gson GSON = new Gson();
        Map<String, Object> json = GSON.fromJson(js, new TypeToken<Map<String, Object>>() {
        }.getType());
        return exArticles(json);
    }

    public String findJsonContent(String html) {
        String st = "msgList = '";
        String et = "}}]}";
        int si = html.indexOf(st);
        if (!(si > -1)) {
            st = "msgList = ";
        }
        si = html.indexOf(st);
        int ei = html.indexOf(et, si);
        if (si > -1 && ei > -1 && ei > si) {
            return html.substring(si + st.length(), ei + et.length());
        } else {
            return "";
        }
    }

    public List<PageTable> exArticles(Map json) {
        LinkedList<PageTable> articles = new LinkedList<>();
        List<Object> items = (List<Object>) json.get("list");
        WeixinAccountConf instance = WeixinAccountConf.getInstance();
        Map<String, AccountTable> accountSeedMap = instance.getAccountMap();
        HashMap<String, String> allBizMap = instance.getBizMap();
        for (Object item : items) {
            Map itemM = (Map) item;
            Object mitemObj = itemM.get("app_msg_ext_info");
            if (mitemObj == null) continue;
            Map mitem = (Map) mitemObj;
            if (((Map) mitemObj).isEmpty()) continue;
            String mtitle = (String) mitem.get("title");
            String murl = (String) mitem.get("content_url");
            if (StringUtils.isEmpty(mtitle) || StringUtils.isEmpty(murl)) continue;
            murl = formatUrl(murl);
            articles.addLast(getPageTable(mtitle, murl, accountSeedMap, allBizMap));

            ArrayList<Object> subitems = (ArrayList<Object>) mitem.get("multi_app_msg_item_list");
            if (CollectionUtils.isEmpty(subitems)) continue;
            for (Object subitem : subitems) {
                Map subitemM = (Map) subitem;
                String subtitle = (String) subitemM.get("title");
                String suburl = (String) subitemM.get("content_url");
                if (StringUtils.isEmpty(subtitle) || StringUtils.isEmpty(suburl)) continue;
                suburl = formatUrl(suburl);
                articles.addLast(getPageTable(subtitle, suburl, accountSeedMap, allBizMap));
            }
        }
        return articles;
    }

    private String formatUrl(String url) {
        return url.replaceAll("&quot;", "\"").replaceAll("&amp;amp;", "&").replace("\\\\/", "/").replace("&nbsp;", " ").replaceAll("&amp;", "&");
    }

    public PageTable getPageTable(String title, String url, Map<String, AccountTable> accountSeedMap, HashMap<String, String> allBizMap) {
        Map<String, String> params = UrlUtil.extractParams(url);
        System.out.println(url);
        String id = StringUtil.genMD5Val(String.format("%s%s%s%s",
                params.get("__biz"),
                params.get("mid"),
                params.get("idx"),
                params.get("sn")));
        String biz = params.get("__biz");
        PageTable table = getPageTableWithAccountIdName(biz, accountSeedMap, allBizMap);
        table.set_id(id);
        table.setPageId(id);
        table.setTitle(title);
        table.setCrawlUrl(url);
        table.setOrgUrl(url);
        return table;
    }

    // get account id and name
    public PageTable getPageTableWithAccountIdName(String biz, Map<String, AccountTable> accountSeedMap, HashMap<String, String> allBizMap) {
        PageTable t = new PageTable();
        String accountId = allBizMap.get(biz);
        String accountName = accountSeedMap.get(accountId).getName();
        if (!StringUtils.isEmpty(accountId)) {
            t.setAccountId(accountId);
        }
        if (!StringUtils.isEmpty(accountName)) {
            t.setAccountName(accountName);
        }
        return t;
    }


}
